package com.gitee.xhs.savePointReader.operatorState;

import cn.hutool.core.date.DateUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;

import java.util.Date;

/**
 * @author BigHead
 * @date 2022/7/4 14:09
 * @description OperatorStateMapFunction
 */
public class OperatorStateMapFunction extends RichMapFunction<Long, OperatorStateExhibitor> {

    private final String jobId;
    private final String operatorUid;
    private final String STATE_TYPE_NAME;
    private Integer taskIndexNum;

    public OperatorStateMapFunction(String jobId, String operatorUid, String stateTypeName) {
        this.jobId = jobId;
        this.operatorUid = operatorUid;
        this.STATE_TYPE_NAME = stateTypeName;
    }

    @Override
    public void open(Configuration parameters) {
        taskIndexNum = 0;
    }

    @Override
    public OperatorStateExhibitor map(Long value) throws Exception {
        OperatorStateExhibitor operatorStateExhibitor = new OperatorStateExhibitor();
        operatorStateExhibitor.setJobId(jobId);
        operatorStateExhibitor.setOperatorUid(operatorUid);
        operatorStateExhibitor.setStateTypeName(STATE_TYPE_NAME);
        operatorStateExhibitor.setTaskIndexNum(taskIndexNum);
        operatorStateExhibitor.setTransactionTimeStamp(value);
        int startYear = DateUtil.beginOfYear(new Date(value)).year();
        operatorStateExhibitor.setTransactionYear(Integer.toString(startYear));

        taskIndexNum += 1;
        return operatorStateExhibitor;
    }
}
